package com.itheima.kafka.boot.stream;

import com.itheima.kafka.boot.config.stream.KafkaStreamListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Arrays;

/**
 * @Description:
 * @Version: V1.0
 */
@Component
@Slf4j
public class StreamListener implements KafkaStreamListener<KStream<String, String>> {

    /**
     * 接收待处理的消息 topic
     * @return
     */
    @Override
    public String listenerTopic() {
        return "itcast-topic-input";
    }

    /**
     * 处理后的结果 发送到新的 topic
     * @return
     */
    @Override
    public String sendTopic() {
        return "itcast-topic-output";
    }

    /**
     * 流式处理逻辑
     * @param stream
     * @return
     */
    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {

       return stream.flatMapValues((key, value) -> {
           return Arrays.asList(value.split(" "));
        })
                .groupBy((key, value) -> {
                    return value;
                })
                .windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
                .count()
                .toStream()
                .map((key,value)->  {
                    System.out.println("key: "+key.key().toString() +" : "+value);
                   return new KeyValue<>(key.key().toString(), value.toString());
                });
    }
}
